package com.asap.demo.mysql.handle;


import com.asap.demo.mysql.orm.DbFetcher;
import com.asap.demo.mysql.table.JsonField;
import com.asap.demo.mysql.table.JsonTable;
import com.asap.demo.mysql.util.PropTransformMap;
import com.asap.demo.mysql.util.TimeUtil;
import com.asap.demo.utils.PluginContext;
import com.asap.demo.utils.Utils;
import com.asap.rule.StandardEvent;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class Process   {
    private static final Logger logger = LoggerFactory.getLogger(Process.class);

    private Map<String, String> configMap = null;
    private String version="V3.0";
    private List<List<Condition>> listCond=null;
    private List<JsonField> updateFields = null;
    private List<JsonField> conditionFields=null;
    private boolean needUpdate = false;
    private String tableName="";
    private String whereCondition="";
    private JsonField primaryKey;
    // 新建一个 DbFetcher 实例
    DbFetcher dbFetcher = null;
    // 获取配置文件中的inputTopic
    RuleParse ruleParse = new RuleParse();

    public boolean initialize(PluginContext Context) throws Exception {
        try {
            logger.info( "当前文件的路径为:{}", Context.getConfigPath());

            // 获得所有配置的映射map
            configMap = PropTransformMap.getInstance().readConfigMap(Context.getConfigPath() + "/conf/cfg.properties");
            ruleParse.setConfigMap(configMap);
            logger.info("configMap"+configMap);

            buildFilter(configMap.get("filter_cond"));
            /**
             * 将自定义配置的 json 反序列化
             */
            ruleParse.parseData(Context.getConfigPath() + "/conf/cfg.json");
            needUpdate = ruleParse.isNeedUpdate();
            updateFields = ruleParse.getUpdateFields();
            conditionFields = ruleParse.getConditionFields();
            tableName = ruleParse.getTableDefine().getTableName();
            if(needUpdate){
                whereCondition = getWhereCondition(ruleParse.getTableDefine().getPrimaryKey());
                primaryKey = ruleParse.getPrimaryKeyField();
            }
            // 新建一个 DbFetcher 实例
            dbFetcher = new DbFetcher(configMap);
        }
        catch (Exception e){
            throw e;
        }

        return true;
    }

    public int processMsg(List<StandardEvent> listEvent) throws Exception {
        if(needUpdate){
            return processNeedUpdateMsg(listEvent);
        }else{
            return processNotNeedUpdateMsg(listEvent);
        }
    }

    public int processNeedUpdateMsg(List<StandardEvent> listEvent) {
        long startTime = System.currentTimeMillis();
        int result = 0;
        logger.info(version + "listEvent.size():" + listEvent.size());
        List<List<String>> listInsertParam = new ArrayList<>();
        Map<String, List<List<String>>> updateParamsMap = new HashMap<>();
        JsonTable table = ruleParse.getTableDefine();
        long insertCnt = 0L;
        for (StandardEvent event : listEvent) {
            try {
                Object parserBody = event.getOriginalBody();
                String jsonStr = parserBody.toString();
                if (!filter(jsonStr)) { // 只处理需要匹配到的数据，在为配置条件是，所有数据都处理
                    logger.info("jsonStr has filter ");
                    continue;
                }

                StandardEvent newEvent = new StandardEvent();
                newEvent = newEvent.parse(jsonStr);

                String key = (String) newEvent.getField(table.getTopicPrimaryKey());
                if (queryData(key)) {// 做update 操作
                    List<String> listParams = new ArrayList<>();
                    String updateSql = createUpdateParam(newEvent, updateFields, listParams);
                    if (updateParamsMap.containsKey(updateSql)) {
                        updateParamsMap.get(updateSql).add(listParams);
                    } else {
                        List<List<String>> list = new ArrayList<>();
                        list.add(listParams);
                        updateParamsMap.put(updateSql, list);
                    }
                } else { // 做insert 操作
                    listInsertParam.add(createParam(newEvent, table));
                    try {
                        dbFetcher.batchInsertMysql(ruleParse.getTableInsertSql(), listInsertParam);
                        insertCnt++;
                        listInsertParam.clear();
                    } catch (Exception e) {
                        logger.error("insert data to database ", e);
                        result = 1;
                    }
                }

            } catch (Exception e) {
                logger.error("error:", e);
                return 1;
            }
        }

        if (!updateParamsMap.isEmpty()) {
            try {
                int updateRecords = 0;
                for (Map.Entry<String, List<List<String>>> ent : updateParamsMap.entrySet()) {
                    updateRecords += ent.getValue().size();
                    dbFetcher.updateMyql(ent.getKey(), ent.getValue());
                }

                logger.info(" 更新记录数：" + updateRecords);
            } catch (Exception e) {
                logger.error("update data to database ", e);
                result = 1;
            }
        }
        /*
        if(listInsertParam.size()>0) {
                try {
                    dbFetcher.batchInsertMysql(ruleParse.getTableInsertSql(), listInsertParam);
                    logger.info(" 入库记录数：" + listInsertParam.size());
                } catch (Exception e) {
                    logger.error( "insert data to database ", e);
                    result = 1;
                }
         }*/
        if (insertCnt > 0) {
            logger.info(" 入库记录数：" + insertCnt);
        }
        long endTime = System.currentTimeMillis();
        logger.info(version + " 入库耗时：" + (endTime - startTime) + "ms ");

        return result;
    }


    public int processNotNeedUpdateMsg(List<StandardEvent> listEvent){
        long startTime = System.currentTimeMillis();
        int result=0;
        logger.info(version+ "listEvent.size():" + listEvent.size());
        List<List<String>> ListParam = new ArrayList<>();

        JsonTable table = ruleParse.getTableDefine();
        for (StandardEvent event : listEvent) {
            try {
                Object parserBody = event.getOriginalBody();
                // 取出需要做判断的源IP和目的IP字段值
                String jsonStr = parserBody.toString();
                if(!filter(jsonStr)){ // 只处理需要匹配到的数据，在为配置条件是，所有数据都处理
                    logger.info("jsonStr has filter ");
                    continue;
                }

                StandardEvent newEvent = new StandardEvent();
                newEvent= newEvent.parse(jsonStr);

                if(StringUtils.equals(event.getHeader().get("topic"),configMap.getOrDefault("dlframe.exceptionTopic","dlExceptionTopic"))){
                    // 异常数据，需要先执行判断是否已插入，如果插入了,就不需要处理
                    String key = (String)newEvent.getField(table.getTopicPrimaryKey());
                    if(queryData(key)){
                        logger.info("data has in database "+ key);
                        continue;
                    }
                }

                ListParam.add(createParam(newEvent, table));
            } catch (Exception e) {
                logger.error("error:", e);
                return 1;
            }
        }

        if(ListParam.size()>0) {
            try {
                dbFetcher.batchInsertMysql(ruleParse.getTableInsertSql(), ListParam);
                logger.info(" 入库记录数：" + ListParam.size());
            } catch (Exception e) {
                logger.error( "insert data to database ", e);
                result = 1;
            }
        }

        long endTime = System.currentTimeMillis();
        logger.info(version+" 入库耗时：" + (endTime - startTime) + "ms ");

        return result;
    }




    /**
     * 创建 sql 参数
     *
     * @param event 标准事件(kafka message)
     * @param table 自定义配置表
     * @return
     */
    public  List<String> createParam(StandardEvent event, JsonTable table) {
        List<String> params = new ArrayList<String>();
        List<JsonField> fields = table.getJsonFieldList();
        // 遍历自定义配置字段集
        for (int i = 0; i < fields.size(); i++) {
            String topicFiledName = fields.get(i).getTopicFieldName();
            String fieldType = fields.get(i).getFieldType();
            String value = null;
            value = String.valueOf(event.getField(topicFiledName));
            String resultValue = "";

            if (value != null && !value.equals("null")) {
                //处理日期字段中的微秒
               if (fields.get(i).getFieldType().equals("date") ) {
                   value = Utils.transforDate(value);
                }
                resultValue = value;
            }
            else {
                if (fields.get(i).getDefaultValue() != null) {
                    // 当字段类型为date且默认值为now或者无默认值时
                    if (StringUtils.equals(fieldType, "date"))
                        resultValue = TimeUtil.getCurrentTime();
                    else
                        resultValue = fields.get(i).getDefaultValue();
                } else {
                    params.add(null);
                }
            }

            params.add(resultValue);
        }

        return params;
    }


    public String getWhereCondition(String primaryKey){
        StringBuffer sb = new StringBuffer();
        if(conditionFields.isEmpty()){
            sb.append(" WHERE ").append(primaryKey).append("=?");
        }else {
            sb.append(" WHERE ");
            int length3 = conditionFields.size();
            for (int j = 0; j < length3; j++) {
                sb.append(conditionFields.get(j).getDbFieldName()).append("=?");
                if (j < length3 - 1) {
                    sb.append(" AND ");
                }
            }
        }

        String updateCondition = configMap.get("update_cond");
        if(!StringUtils.isEmpty(updateCondition)){
            sb.append("  AND ").append(updateCondition);
        }

        return sb.toString();
    }

    public  String createUpdateParam(StandardEvent event, List<JsonField> fields,List<String> params){
        // 遍历自定义配置字段集
        StringBuffer sb = new StringBuffer();
        sb.append("UPDATE ").append(tableName).append(" SET ");
        for (int i = 0; i < fields.size(); i++) {
            String topicFiledName = fields.get(i).getTopicFieldName();
            String dbFieldName = fields.get(i).getDbFieldName();
            String value = "";
            if(event.getField(topicFiledName)==null){
                continue;
            }else{
                if(fields.get(i).isNeedUpdateAdd()){
                    value =  String.valueOf(event.getField(topicFiledName));
                    sb.append(dbFieldName).append("=").append(dbFieldName).append("+? ,");
                    params.add(value);
                }else{
                    value =  String.valueOf(event.getField(topicFiledName));
                    if (fields.get(i).getFieldType().equals("date") ) {
                        value = Utils.transforDate(value);
                    }
                    sb.append(dbFieldName).append("=? ,");
                    params.add(value);
                }


            }
        }

        params.add((String)event.getField(primaryKey.getTopicFieldName()));

       String sql = sb.substring(0, sb.length()-2) + whereCondition;
        return sql;
    }

    private boolean queryData(String keys){
        return queryMysqlData(keys);
    }

    private boolean queryMysqlData(String key){
        String sql = ruleParse.getTableQuerySql();
        sql = sql.replace("?", key);
       return dbFetcher.queryMysql(sql);
    }


    private Condition parseCondition(String cond){
        // 在对条件进行解析的时候， 因为匹配的原始数据都是json的，所以在条件上都加了""还有:
        // 比如配置的条件为DEVICE_PARENT_TYPE=IPS, 实际数据中的是 "DEVICE_PARENT_TYPE":"IPS",所以在构建最终用于contain匹配时的条件
        //有将原始配置的条件，加上""，且key,value之间加上了:
        StringBuffer sb = new StringBuffer();
        Condition condition = null;
        if(cond.contains("!=")){
            String[] elems = cond.split("!=");
            sb.append("\"").append(elems[0].trim()).append("\"").append(":\"").append(elems[1].trim()).append("\"");
            condition = new Condition(1,sb.toString());
        }else{
            String[] elems = cond.split("=");
            sb.append("\"").append(elems[0].trim()).append("\"").append(":\"").append(elems[1].trim()).append("\"");
            condition = new Condition(0, sb.toString());
        }

        return condition;
    }

    private List<Condition> parseCondGroups(String condGroups){
        List<Condition> listCondGroups = new ArrayList<>();
        if(StringUtils.isNotEmpty(condGroups)){
            if(condGroups.contains(" AND ")){
                String[] conds = condGroups.split(" AND ");
                for(int i=0; i<conds.length; i++){
                    if(StringUtils.isNotEmpty(conds[i])) {
                        listCondGroups.add(parseCondition(conds[i]));
                    }
                }
            }else{
                listCondGroups.add( parseCondition(condGroups));
            }
        }

        return listCondGroups;
    }

    private void buildFilter(String filterCond){
        this.listCond = new ArrayList<>();
        if(StringUtils.isNotEmpty(filterCond)){
            if(filterCond.contains(" OR ")){
                String[] condGroups = filterCond.split(" OR ");
                for(int i=0; i< condGroups.length; i++){
                    listCond.add(parseCondGroups(condGroups[i]));
                }
            }else{
                listCond.add(parseCondGroups(filterCond));
            }
        }

        logger.info("listCond.size:"+listCond.size());
        logger.info("listCond"+listCond.toString());
    }

    private  boolean matchCond(Condition cond, String json){
        boolean flag = true;
        if(cond.getType()==0){ // 0 正常匹配
            if (!json.contains(cond.getCondition())){  //没匹配上条件 比如 DEVICE_PARENT_TYPE=IPS,数据中是DEVICE_PARENT_TYPE=WAF
               flag = false;
            }
        }else{ //1 取反匹配
            if (json.contains(cond.getCondition())){ //没匹配上条件，比如配置的条件为EVENT_ONE_TYPE!=99, 数据中是 EVENT_ONE_TYPE=99
                flag = false;
            }
        }
        //logger.info("cond"+cond.toString());
        return  flag;
    }

    private boolean matchConds(List<Condition>conds,String json){
        for(Condition cond:conds){
            if(!matchCond(cond,json)){ // 在一个条件组里面，一个条件没匹配上，就是没匹配上
                return false;
            }
        }

        return  true;
    }

    private boolean filter(String json){
       // logger.info("json"+json);
        if(this.listCond.isEmpty()) {
            return true;
        }else{
            for(List<Condition>conds:listCond){
                if(matchConds(conds,json)){ // 有一个条件组匹配，就表示匹配上了
                    return true;
                }
            }
        }
        return false;
    }
}

